package db.prob.emp;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Task.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import db.prob.operators.join.MapperJoin;
import db.prob.operators.join.ReducerJoin;
import db.prob.operators.project.MapperProject;
import db.prob.operators.project.ReducerProject;

public class Suciu extends Configured implements Tool {
	private static final Log LOG = LogFactory.getLog("Suciu");
	
	// Input tables
	Path emp;
	Path dept;
	Path rank;
	
	// Temporary tables
	Path r1;
	Path r2;
	Path r3;
	Path r4;
	
	public Suciu(String inputDir) {
		this.emp  = new Path(inputDir + "emp.txt");
		this.dept = new Path(inputDir + "dept.txt");
		this.rank = new Path(inputDir + "rank.txt");
		
		this.r1 = new Path(inputDir + "suciu/r1");
		this.r2 = new Path(inputDir + "suciu/r2");
		this.r3 = new Path(inputDir + "suciu/r3");
		this.r4 = new Path(inputDir + "suciu/r4");

	}

	public static long jobCommCost (Job j) {
		try {
			return 
				j.getCounters().findCounter(Counter.MAP_INPUT_RECORDS).getValue() + 
				j.getCounters().findCounter(Counter.REDUCE_INPUT_RECORDS).getValue();
		} catch (IOException e) {
			LOG.error("Problem with counter.");
			e.printStackTrace();
			return 0;
		}		
	}

	@Override
	public int run(String[] arg0) throws Exception {
		long numOfMessages = 0;
		long startTime = System.currentTimeMillis();				
		numOfMessages += projectEmp();
		numOfMessages += joinR1withD();
		numOfMessages += joinR2withR();
		numOfMessages += projectR3();
		long stopTime = System.currentTimeMillis();
		
		LOG.info(String.format("Total communication cost is: %d messages.", numOfMessages));
		LOG.info(String.format("Wall clock time is: %d milliseconds.", stopTime - startTime));
		return 0;
	}

	private long projectR3() throws Exception {
		Configuration conf = this.getConf();
		
		conf.set(MapperProject.TABLE_NAME,"r3");
		conf.set(MapperProject.COLUMNS,"0,1");
		conf.set(MapperProject.RESULT_TABLE,"result");
		
		Job job = new Job(conf);
		job.setJobName("Project_{rid,did}(r3)");
		
		job.setMapperClass(MapperProject.class);
        job.setReducerClass(ReducerProject.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(r4)) {
			fs.delete(r4, true);
		}

		FileInputFormat.setInputPaths(job, r3);
		FileOutputFormat.setOutputPath(job, r4);
		
		job.waitForCompletion(true);
		return jobCommCost(job);		

	}

	private long joinR2withR() throws Exception {
		Configuration conf = this.getConf();
		
		conf.set(MapperJoin.TABLE_1, "r2");
		conf.setInt(MapperJoin.TABLE_1_ATTR, 1); // sid
		conf.set(MapperJoin.TABLE_2, "rank");
		conf.setInt(MapperJoin.TABLE_2_ATTR, 0); // sid
		conf.set(MapperJoin.RESULT_TABLE, "r3");
		
		Job job = new Job(conf);
		job.setJobName("Join r2 with rank");

		job.setMapperClass(MapperJoin.class);
        job.setReducerClass(ReducerJoin.class);
		
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

		FileInputFormat.setInputPaths(job, r2, rank);
		FileOutputFormat.setOutputPath(job, r3);
        FileSystem fs = FileSystem.get(conf);

        if (fs.exists(r3)) {
			fs.delete(r3, true);
		}
		
		job.waitForCompletion(true);
		
		return jobCommCost(job);
	}

	private long joinR1withD() throws Exception {
		Configuration conf = this.getConf();
		
		conf.set(MapperJoin.TABLE_1, "r1");
		conf.setInt(MapperJoin.TABLE_1_ATTR, 0); // sid
		conf.set(MapperJoin.TABLE_2, "dept");
		conf.setInt(MapperJoin.TABLE_2_ATTR, 0); // sid
		conf.set(MapperJoin.RESULT_TABLE, "r2");
		
		Job job = new Job(conf);
		job.setJobName("Join r1 with dept");

		job.setMapperClass(MapperJoin.class);
        job.setReducerClass(ReducerJoin.class);
		
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

		FileInputFormat.setInputPaths(job, r1, dept);
		FileOutputFormat.setOutputPath(job, r2);
        FileSystem fs = FileSystem.get(conf);

        if (fs.exists(r2)) {
			fs.delete(r2, true);
		}
		
		job.waitForCompletion(true);
		
		return jobCommCost(job);
	}

	private long projectEmp() throws Exception {
		Configuration conf = this.getConf();
		
		conf.set(MapperProject.TABLE_NAME,"emp");
		conf.set(MapperProject.COLUMNS,"2,3");
		conf.set(MapperProject.RESULT_TABLE,"r1");
		
		Job job = new Job(conf);
		job.setJobName("Project_{rid,did}(emp)");
		
		job.setMapperClass(MapperProject.class);
        job.setReducerClass(ReducerProject.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(r1)) {
			fs.delete(r1, true);
		}

		FileInputFormat.setInputPaths(job, emp);
		FileOutputFormat.setOutputPath(job, r1);
		
		job.waitForCompletion(true);
		return jobCommCost(job);		
	}

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		String inputDir = args[0];
		int ret = ToolRunner.run(new Suciu(inputDir), args);
		System.exit(ret);
	}

}
